package com.thiswhyme.stream.rabbit.consumer2.service;

import com.thiswhyme.stream.rabbit.consumer2.BizService.InputChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

/**
 * @description:
 * @author: thiswhyme
 * @date: 2020/9/16
 */
@Slf4j
@Service
public class BizService {

    @StreamListener(InputChannel.INPUT_CHANNEL_LOG)
    public void onLogMessage(String msg) {
        log.info("【consumer2】监听到INPUT_CHANNEL_LOG消息，内容：{}", msg);
//        throw new RuntimeException("失败");
    }

    @StreamListener(InputChannel.INPUT_CHANNEL_LOG2)
    public void onLog2Message(String msg) {
        log.info("【consumer2】监听到INPUT_CHANNEL_LOG2消息，内容：{}", msg);
    }

    @StreamListener(InputChannel.INPUT_CHANNEL_DELAY)
    public void onDelayMessage(String msg) {
        log.info("【consumer2】监听到INPUT_CHANNEL_DELAY消息，内容：{}", msg);
    }

//    @StreamListener(InputChannel.INPUT_CHANNEL_MANUAL)
//    public void onManualMessage(String msg, @Header(AmqpHeaders.CHANNEL) Channel channel, @Headers
//                                @Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag) throws IOException {
//        log.info("【consumer2】监听到INPUT_CHANNEL_MANUAL消息，内容：{}", msg);
//        if("false".equals(msg)) {
//            channel.basicAck(deliveryTag, false);
//        } else {
//            channel.basicNack(deliveryTag,false,false);
//        }
//    }

//    @StreamListener(InputChannel.INPUT_CHANNEL_MANUAL)
//    public void onManualMessage(Message<String> message) throws IOException {
//        int retries = (int) Optional.ofNullable(message.getHeaders().get("RETRIES_HEADER")).orElse(0);
//        log.info("【consumer2】监听到INPUT_CHANNEL_MANUAL消息，内容：" + message.getPayload() + " 第【" + retries + "】次重试");
//        Channel channel = (Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
//        Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
//
//        if("true".equals(message.getPayload())) {
//            log.info("【consumer2】INPUT_CHANNEL_MANUAL消息，消息被消费，内容：" + message.getPayload());
//            channel.basicAck(deliveryTag, false);
//        } else if("true1".equals(message.getPayload())) {
//            channel.basicAck(deliveryTag, false);
//            if (retries < 3){
//                Map<String, Object> headerMap = new HashMap<>();
//                headerMap.put("RETRIES_HEADER", retries + 1);
//                AMQP.BasicProperties basicProperties = new AMQP.BasicProperties("application/json", (String)null, headerMap, 2, 0, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);
//
//                channel.basicPublish((String) message.getHeaders().get(AmqpHeaders.RECEIVED_EXCHANGE),
//                        (String) message.getHeaders().get(AmqpHeaders.RECEIVED_ROUTING_KEY), basicProperties,
//                        message.getPayload().getBytes());
//            } else {
//                log.info("【consumer2】INPUT_CHANNEL_MANUAL消息，重试3次后仍然失败，内容：" + message.getPayload());
//            }
//        } else {
//            log.info("【consumer2】INPUT_CHANNEL_MANUAL消息，消息被拒绝，内容：" + message.getPayload());
//            channel.basicReject(deliveryTag, false);
//        }
//    }

    @StreamListener(InputChannel.INPUT_CHANNEL_TOPIC_ROUTE1)
    public void onTopicRoute1Message(String msg) {
        log.info("【consumer2】监听到INPUT_CHANNEL_TOPIC_ROUTE1消息，内容：{}", msg);
    }

    @StreamListener(InputChannel.INPUT_CHANNEL_TOPIC_ROUTE2)
    public void onTopicRoute2Message(String msg) {
        log.info("【consumer2】监听到INPUT_CHANNEL_TOPIC_ROUTE2消息，内容：{}", msg);
    }

    @StreamListener(InputChannel.INPUT_CHANNEL_CONFIRM)
    public void onTConfirmMessage(Message<String> msg) {
        log.info("【consumer2】监听到INPUT_CHANNEL_CONFIRM消息，{}内容：{}",
                null == msg.getHeaders().get("biz-republish-times") ? "" : "第【" + msg.getHeaders().get("biz-republish-times") + "】次重发，", msg.getPayload());
    }
}
